package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 指数分时行情
 */
public class IndexMinWindowFunction extends RichWindowFunction<CleanBean, IndexBean,String, TimeWindow> {
    /**
     * 开发步骤：
     * 1.初始化 MapState<String, IndexBean>
     * 2.记录最新指数
     * 3.获取分时成交额和成交数量
     * 4.格式化日期
     * 5.封装输出数据
     * 6.更新MapState
     */
    MapState<String, IndexBean> stockMs = null;
    //1.初始化 MapState<String, StockBean>
    //通过mapState缓存当前最新个股行情,供下次计算分时成交量/金额使用
    @Override
    public void open(Configuration parameters) throws Exception {
        stockMs = getRuntimeContext().getMapState(new MapStateDescriptor<String, IndexBean>("stockMs", String.class, IndexBean.class));
    }

    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<IndexBean> out) throws Exception {

        //2.记录最新个股
        CleanBean line = null; //每分钟,第59s的数据
        for (CleanBean cleanBean : input) {
            if (line == null) {
                line = cleanBean;
            }
            if (line.getEventTime() < cleanBean.getEventTime()) {
                line = cleanBean;
            }
        }
        //3.获取分时成交额和成交数量
        //现获取上一分钟缓存的个股分时行情数据
        IndexBean stockBeanLast = stockMs.get(line.getSecCode());
        Long minVol = 0L; //分时成交量
        Long minAmt = 0L; //分时成交金额
        if (stockBeanLast != null) {
            Long tradeVolDayLast = stockBeanLast.getTradeVolDay(); //上一分钟的日成交总量
            Long tradeAmtDayLast = stockBeanLast.getTradeAmtDay();//上一分钟的日成交金额
            minAmt = line.getTradeAmt() - tradeAmtDayLast;
            minVol = line.getTradeVolumn() - tradeVolDayLast;
        }


        //4.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        String tradeDate = DateUtil.longTimeToString(line.getEventTime(), Constant.format_yyyy_mm_dd);

        //5.封装输出数据
        IndexBean indexBeanNew = new IndexBean(
                tradeTime,
                line.getSecCode(),
                line.getSecName(),
                line.getPreClosePrice(),
                line.getOpenPrice(),
                line.getHighPrice(),
                line.getLowPrice(),
                line.getTradePrice(),
                minVol, minAmt,
                line.getTradeVolumn(),
                line.getTradeAmt(),
                tradeDate,
                line.getEventTime(),
                line.getSource()
        );
        out.collect(indexBeanNew);
        //6.更新MapState
        stockMs.put(indexBeanNew.getIndexCode(), indexBeanNew);
    }
}
